package queue

import (
	"fmt"

	"github.com/streadway/amqp"
	"golang-manage-admin/providers"
)

/**
 * 新增一个 rabbitmq 管理
 */
func NewRabbitMq(exchangeName, exchangeType, routingKey, queueName string, qos int, region string, args amqp.Table, isDeadLetter bool) *RabbitMq {
	connection, channel := providers.NewRabbitmq(region)

	return &RabbitMq{
		isDeadLetter: isDeadLetter,
		exchangeName: exchangeName,
		exchangeType: exchangeType,
		routingKey:   routingKey,
		queueName:    queueName,
		qos:          qos,
		args:         args,

		mqConnection: connection,
		mqChannel:    channel,
	}
}

// rabbitmq 队列处理
type RabbitMq struct {
	isDeadLetter bool
	exchangeName string
	exchangeType string
	routingKey   string
	queueName    string
	qos          int

	args         amqp.Table
	mqConnection *amqp.Connection // rabbitmq 连接
	mqChannel    *amqp.Channel    // rabbitmq 的通道
}

/**
 *  @Author yang
 *  @Description 队列绑定
**/
func (mq *RabbitMq) Bind() <-chan amqp.Delivery {
	/**
	 * prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
	 * prefetchSize：0
	 * global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
	 */
	// 消息数量限定, 避免消息堵塞
	mq.mqChannel.Qos(mq.qos, 0, false)

	// 1.交换机声明
	var exchangeArgs amqp.Table
	if mq.exchangeType == "x-delayed-message" {
		exchangeArgs = amqp.Table{"x-delayed-type": "direct"}
	}
	err := mq.mqChannel.ExchangeDeclare(
		mq.exchangeName, //交换器名称,
		mq.exchangeType, //exchange type：一般用fanout、direct、topic
		true,            // 是否持久化
		false,           //是否自动删除（自动删除的前提是至少有一个队列或者交换器与这和交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑）
		false,           //设置是否内置的。true表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式
		false,           //是否阻塞
		exchangeArgs,    // 辅助属性
	)

	if err != nil {
		errInfo := fmt.Sprintf("[%s]交换机声明出错:%s", mq.exchangeName, err.Error())
		providers.Logger.Error(errInfo)
		panic(errInfo)
	}

	// 2.队列声明
	_, err = mq.mqChannel.QueueDeclare(
		mq.queueName, // 队列名称
		true,         // 是否持久化
		false,        // 是否自动删除 (前提是至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。注意：生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列)
		false,        // 是否为排他队列（排他的队列仅对“首次”声明的conn可见[一个conn中的其他channel也能访问该队列]，conn结束后队列删除）
		false,        // 是否阻塞
		mq.args,
	)

	if err != nil {
		errInfo := fmt.Sprintf("[%s]队列声明出错:%s", mq.queueName, err.Error())
		providers.Logger.Error(errInfo)
		panic(errInfo)
	}

	// 3.建立绑定 Binding
	err = mq.mqChannel.QueueBind(
		mq.queueName,    //name 队列名称
		mq.routingKey,   //key
		mq.exchangeName, //exchange string 交换机名称
		false,           //noWait bool
		nil,             //args Table
	)

	if err != nil {
		errInfo := fmt.Sprintf("[%s]队列与[%s]交换机绑定时出错:%s", mq.queueName, mq.exchangeName, err.Error())
		providers.Logger.Errorf(errInfo)
		panic(errInfo)
	}

	// 2.从队列中获取消息（customer 只需关注队列信息即可）
	var deliveryChan <-chan amqp.Delivery
	if !mq.isDeadLetter {
		deliveryChan, err = mq.mqChannel.Consume(
			mq.queueName, // 队列名称
			"",           //consumer string
			false,        //autoAck
			false,        //exclusive
			false,        //noLocal
			false,        //noWait bool
			nil,
		)
	}

	if err != nil {
		errInfo := fmt.Sprintf("[%s]队列监听获取的通道有误:%s", mq.queueName, err.Error())
		providers.Logger.Errorf(errInfo)
		panic(errInfo)
	}

	providers.Logger.Infof("开启,%+v", mq)

	return deliveryChan
}

/**
 * mq 关闭
 */
func (mq *RabbitMq) Close() {
	if mq.mqChannel != nil {
		mq.mqChannel.Close()
	}

	if mq.mqConnection != nil {
		mq.mqConnection.Close()
	}
}
